﻿using System.Collections.Generic;

namespace System.Linq
{
    public static class Observable_Join
    {
        public static IObservable<Tuple<TLeft, TRight>> Join<TLeft, TRight>(this IObservable<TLeft> left, IObservable<TRight> right)
        {
            return left.Zip(right, (l, r) => new Tuple<TLeft, TRight>(l, r));
            //Note: Ok, as it turns out, Join duplicates Zip. But nevertheless it was an interesting experience to play around with Rx like this.
            //return new ThreadsafeJoinedObservable<TLeft, TRight>(left, right);
        }


        private class ThreadsafeJoinedObservable<TLeft, TRight> : IObservable<Tuple<TLeft, TRight>>
        {
            private readonly Subject<Tuple<TLeft, TRight>> joiner = new Subject<Tuple<TLeft, TRight>>();

            private readonly Queue<TLeft> leftItems = new Queue<TLeft>();
            private readonly Queue<TRight> rightItems = new Queue<TRight>();


            public ThreadsafeJoinedObservable(IObservable<TLeft> left, IObservable<TRight> right)
            {
                left.Subscribe(item => Process(item, this.leftItems), ex => this.joiner.OnError(ex), () => this.joiner.OnCompleted());
                right.Subscribe(item => Process(item, this.rightItems), ex => this.joiner.OnError(ex), () => this.joiner.OnCompleted());
            }


            private void Process<T>(T item, Queue<T> items)
            {
                lock(this.joiner)
                {
                    items.Enqueue(item);

                    if (this.leftItems.Count > 0 && this.rightItems.Count > 0)
                        this.joiner.OnNext(new Tuple<TLeft, TRight>(this.leftItems.Dequeue(),
                                                                    this.rightItems.Dequeue()));
                }
            }


            public IDisposable Subscribe(IObserver<Tuple<TLeft, TRight>> observer)
            {
                return this.joiner.Subscribe(observer);
            }
        }
    }
}
